// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/type/type.h>

#include <boost/algorithm/string.hpp>
#include <fmt/format.h>
#include <melon/demangle.h>
#include <re2/re2.h>

#include <sstream>
#include <typeindex>
#include <melon/container/f14_map.h>
#include <pollux/type/timestamp_conversion.h>

namespace std {
    template<>
    struct hash<kumo::pollux::TypeKind> {
        size_t operator()(const kumo::pollux::TypeKind &typeKind) const {
            return std::hash<int32_t>()((int32_t) typeKind);
        }
    };
} // namespace std

namespace {
    bool isColumnNameRequiringEscaping(const std::string &name) {
        static const std::string re("^[a-zA-Z_][a-zA-Z0-9_]*$");
        return !RE2::FullMatch(name, re);
    }
} // namespace

namespace kumo::pollux {
    // Static variable intialization is not thread safe for non
    // constant-initialization, but scoped static initialization is thread safe.
    const melon::F14FastMap<std::string, TypeKind> &getTypeStringMap() {
        static const melon::F14FastMap<std::string, TypeKind> kTypeStringMap{
            {"BOOLEAN", TypeKind::BOOLEAN},
            {"TINYINT", TypeKind::TINYINT},
            {"SMALLINT", TypeKind::SMALLINT},
            {"INTEGER", TypeKind::INTEGER},
            {"BIGINT", TypeKind::BIGINT},
            {"HUGEINT", TypeKind::HUGEINT},
            {"REAL", TypeKind::REAL},
            {"DOUBLE", TypeKind::DOUBLE},
            {"VARCHAR", TypeKind::VARCHAR},
            {"VARBINARY", TypeKind::VARBINARY},
            {"TIMESTAMP", TypeKind::TIMESTAMP},
            {"ARRAY", TypeKind::ARRAY},
            {"MAP", TypeKind::MAP},
            {"ROW", TypeKind::ROW},
            {"FUNCTION", TypeKind::FUNCTION},
            {"UNKNOWN", TypeKind::UNKNOWN},
            {"OPAQUE", TypeKind::OPAQUE},
            {"INVALID", TypeKind::INVALID}
        };
        return kTypeStringMap;
    }

    std::optional<TypeKind> tryMapNameToTypeKind(const std::string &name) {
        auto found = getTypeStringMap().find(name);

        if (found == getTypeStringMap().end()) {
            return std::nullopt;
        }

        return found->second;
    }

    TypeKind mapNameToTypeKind(const std::string &name) {
        auto found = getTypeStringMap().find(name);

        if (found == getTypeStringMap().end()) {
            POLLUX_USER_FAIL("Specified element is not found : {}", name);
        }

        return found->second;
    }

    std::string mapTypeKindToName(const TypeKind &typeKind) {
        static melon::F14FastMap<TypeKind, std::string> typeEnumMap{
            {TypeKind::BOOLEAN, "BOOLEAN"},
            {TypeKind::TINYINT, "TINYINT"},
            {TypeKind::SMALLINT, "SMALLINT"},
            {TypeKind::INTEGER, "INTEGER"},
            {TypeKind::BIGINT, "BIGINT"},
            {TypeKind::HUGEINT, "HUGEINT"},
            {TypeKind::REAL, "REAL"},
            {TypeKind::DOUBLE, "DOUBLE"},
            {TypeKind::VARCHAR, "VARCHAR"},
            {TypeKind::VARBINARY, "VARBINARY"},
            {TypeKind::TIMESTAMP, "TIMESTAMP"},
            {TypeKind::ARRAY, "ARRAY"},
            {TypeKind::MAP, "MAP"},
            {TypeKind::ROW, "ROW"},
            {TypeKind::FUNCTION, "FUNCTION"},
            {TypeKind::UNKNOWN, "UNKNOWN"},
            {TypeKind::OPAQUE, "OPAQUE"},
            {TypeKind::INVALID, "INVALID"}
        };

        auto found = typeEnumMap.find(typeKind);

        if (found == typeEnumMap.end()) {
            POLLUX_USER_FAIL("Specified element is not found : {}", (int32_t)typeKind);
        }

        return found->second;
    }

    std::pair<uint8_t, uint8_t> getDecimalPrecisionScale(const Type &type) {
        if (type.isShortDecimal()) {
            const auto &decimalType = static_cast<const ShortDecimalType &>(type);
            return {decimalType.precision(), decimalType.scale()};
        } else if (type.isLongDecimal()) {
            const auto &decimalType = static_cast<const LongDecimalType &>(type);
            return {decimalType.precision(), decimalType.scale()};
        }
        POLLUX_FAIL("Type is not Decimal");
    }

    namespace {
        struct OpaqueSerdeRegistry {
            struct Entry {
                std::string persistentName;
                // to avoid creating new shared_ptr's every time
                OpaqueType::SerializeFunc<void> serialize;
                OpaqueType::DeserializeFunc<void> deserialize;
            };

            melon::F14FastMap<std::type_index, Entry> mapping;
            melon::F14FastMap<std::string, std::shared_ptr<const OpaqueType> > reverse;

            static OpaqueSerdeRegistry &get() {
                static OpaqueSerdeRegistry instance;
                return instance;
            }
        };
    } // namespace

    std::ostream &operator<<(std::ostream &os, const TypeKind &kind) {
        os << mapTypeKindToName(kind);
        return os;
    }

    namespace {
        std::vector<TypePtr> deserializeChildTypes(const melon::Dynamic &obj) {
            return pollux::ISerializable::deserialize<std::vector<Type> >(obj["cTypes"]);
        }
    } // namespace

    TypePtr Type::create(const melon::Dynamic &obj) {
        if (obj.find("ref") != obj.items().end()) {
            const auto id = obj["ref"].asInt();
            return deserializedTypeCache().get(id);
        }

        std::vector<TypePtr> childTypes;
        if (obj.find("cTypes") != obj.items().end()) {
            childTypes = deserializeChildTypes(obj);
        }

        auto typeName = obj["type"].asString();
        if (isDecimalName(typeName)) {
            return DECIMAL(obj["precision"].asInt(), obj["scale"].asInt());
        }
        // Checks if 'typeName' specifies a custom type.
        if (customTypeExists(typeName)) {
            std::vector<TypeParameter> params;
            params.reserve(childTypes.size());
            for (auto &child: childTypes) {
                params.emplace_back(child);
            }
            return getCustomType(typeName, params);
        }

        // 'typeName' must be a built-in type.
        TypeKind typeKind = mapNameToTypeKind(typeName);
        switch (typeKind) {
            case TypeKind::ROW: {
                POLLUX_USER_CHECK(obj["names"].isArray());
                std::vector<std::string> names;
                for (const auto &name: obj["names"]) {
                    names.push_back(name.asString());
                }

                return std::make_shared<const RowType>(
                    std::move(names), std::move(childTypes));
            }

            case TypeKind::OPAQUE: {
                const auto &persistentName = obj["opaque"].asString();
                const auto &registry = OpaqueSerdeRegistry::get();
                auto it = registry.reverse.find(persistentName);
                POLLUX_USER_CHECK(
                    it != registry.reverse.end(),
                    "Opaque type with persistent name '{}' is not registered",
                    persistentName);
                if (auto withExtra = it->second->deserializeExtra(obj)) {
                    return withExtra;
                }
                return it->second;
            }
            default: {
                return createType(typeKind, std::move(childTypes));
            }
        }
    }

    // static
    void Type::registerSerDe() {
        auto &registry = pollux::DeserializationRegistryForSharedPtr();
        registry.Register(
            Type::getClassName(),
            static_cast<TypePtr (*)(const melon::Dynamic &)>(Type::create));

        registry.Register("IntervalDayTimeType", IntervalDayTimeType::deserialize);
        registry.Register(
            "IntervalYearMonthType", IntervalYearMonthType::deserialize);
        registry.Register("DateType", DateType::deserialize);
    }

    std::string ArrayType::toString() const {
        return "ARRAY<" + child_->toString() + ">";
    }

    const TypePtr &ArrayType::childAt(uint32_t idx) const {
        POLLUX_USER_CHECK_EQ(idx, 0, "Array type should have only one child");
        return elementType();
    }

    ArrayType::ArrayType(TypePtr child)
        : child_{std::move(child)}, parameters_{{TypeParameter(child_)}} {
    }

    bool ArrayType::equivalent(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        auto &otherArray = other.as_array();
        return child_->equivalent(*otherArray.child_);
    }

    bool ArrayType::equals(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        auto &otherArray = other.as_array();
        return *child_ == *otherArray.child_;
    }

    melon::Dynamic ArrayType::serialize() const {
        melon::Dynamic obj = melon::Dynamic::object;
        obj["name"] = "Type";
        obj["type"] = TypeTraits<TypeKind::ARRAY>::name;

        melon::Dynamic children = melon::Dynamic::array;
        children.push_back(child_->serialize());
        obj["cTypes"] = children;

        return obj;
    }

    const TypePtr &MapType::childAt(uint32_t idx) const {
        if (idx == 0) {
            return keyType();
        } else if (idx == 1) {
            return valueType();
        }
        POLLUX_USER_FAIL(
            "Map type should have only two children. Tried to access child '{}'",
            idx);
    }

    const char *MapType::nameOf(uint32_t idx) const {
        if (idx == 0) {
            return "key";
        } else if (idx == 1) {
            return "value";
        }
        POLLUX_USER_FAIL(
            "Map type should have only two children. Tried to get name of child '{}'",
            idx);
    }

    MapType::MapType(TypePtr keyType, TypePtr valueType)
        : keyType_{std::move(keyType)},
          valueType_{std::move(valueType)},
          parameters_{{TypeParameter(keyType_), TypeParameter(valueType_)}} {
    }

    std::string MapType::toString() const {
        return "MAP<" + keyType()->toString() + "," + valueType()->toString() + ">";
    }

    melon::Dynamic MapType::serialize() const {
        melon::Dynamic obj = melon::Dynamic::object;
        obj["name"] = "Type";
        obj["type"] = TypeTraits<TypeKind::MAP>::name;

        melon::Dynamic children = melon::Dynamic::array;
        children.push_back(keyType()->serialize());
        children.push_back(valueType()->serialize());
        obj["cTypes"] = children;

        return obj;
    }

    namespace {
        std::vector<TypeParameter> createTypeParameters(
            const std::vector<TypePtr> &children) {
            std::vector<TypeParameter> parameters;
            parameters.reserve(children.size());
            for (const auto &child: children) {
                parameters.push_back(TypeParameter(child));
            }
            return parameters;
        }

        std::string namesAndTypesToString(
            const std::vector<std::string> &names,
            const std::vector<TypePtr> &types) {
            std::stringstream ss;
            ss << "[names: {";
            if (!names.empty()) {
                for (const auto &name: names) {
                    ss << "'" << name << "', ";
                }
                ss.seekp(-2, std::ios_base::cur);
            } else {
                ss << " ";
            }
            ss << "}, types: {";
            if (!types.empty()) {
                for (const auto &type: types) {
                    ss << (type ? type->toString() : "NULL") << ", ";
                }
                ss.seekp(-2, std::ios_base::cur);
            } else {
                ss << " ";
            }
            ss << "}]";
            return ss.str();
        }
    } // namespace

    RowType::RowType(std::vector<std::string> &&names, std::vector<TypePtr> &&types)
        : names_{std::move(names)}, children_{std::move(types)} {
        POLLUX_CHECK_EQ(
            names_.size(),
            children_.size(),
            "Mismatch names/types sizes: {}",
            namesAndTypesToString(names_, children_));
        for (auto &child: children_) {
            POLLUX_CHECK_NOT_NULL(
                child,
                "Child types cannot be null: {}",
                namesAndTypesToString(names_, children_));
        }
    }

    RowType::~RowType() {
        if (auto *parameters = parameters_.load()) {
            delete parameters;
        }
    }

    std::unique_ptr<std::vector<TypeParameter> > RowType::makeParameters() const {
        return std::make_unique<std::vector<TypeParameter> >(
            createTypeParameters(children_));
    }

    namespace {
        template<typename T>
        std::string makeFieldNotFoundErrorMessage(
            const T &name,
            const std::vector<std::string> &availableNames) {
            std::stringstream errorMessage;
            errorMessage << "Field not found: " << name << ". Available fields are: ";
            for (auto i = 0; i < availableNames.size(); ++i) {
                if (i > 0) {
                    errorMessage << ", ";
                }
                errorMessage << availableNames[i];
            }
            errorMessage << ".";
            return errorMessage.str();
        }
    } // namespace

    const TypePtr &RowType::findChild(melon::StringPiece name) const {
        for (uint32_t i = 0; i < names_.size(); ++i) {
            if (names_.at(i) == name) {
                return children_.at(i);
            }
        }
        POLLUX_USER_FAIL(makeFieldNotFoundErrorMessage(name, names_));
    }

    bool RowType::isOrderable() const {
        return std::all_of(
            children_.cbegin(), children_.cend(), [](const auto &child) {
                return child->isOrderable();
            });
    }

    bool RowType::isComparable() const {
        return std::all_of(
            children_.cbegin(), children_.cend(), [](const auto &child) {
                return child->isComparable();
            });
    }

    bool RowType::containsChild(std::string_view name) const {
        return std::find(names_.begin(), names_.end(), name) != names_.end();
    }

    uint32_t RowType::getChildIdx(std::string_view name) const {
        auto index = getChildIdxIfExists(name);
        if (!index.has_value()) {
            POLLUX_USER_FAIL(makeFieldNotFoundErrorMessage(name, names_));
        }
        return index.value();
    }

    std::optional<uint32_t> RowType::getChildIdxIfExists(
        std::string_view name) const {
        for (uint32_t i = 0; i < names_.size(); i++) {
            if (names_.at(i) == name) {
                return i;
            }
        }
        return std::nullopt;
    }

    bool RowType::equivalent(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        const auto &otherTyped = other.as_row();
        if (otherTyped.size() != size()) {
            return false;
        }
        for (size_t i = 0; i < size(); ++i) {
            if (!childAt(i)->equivalent(*otherTyped.childAt(i))) {
                return false;
            }
        }
        return true;
    }

    bool RowType::equals(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        const auto &otherTyped = other.as_row();
        if (otherTyped.size() != size()) {
            return false;
        }
        for (size_t i = 0; i < size(); ++i) {
            // todo: case sensitivity
            if (nameOf(i) != otherTyped.nameOf(i) ||
                *childAt(i) != *otherTyped.childAt(i)) {
                return false;
            }
        }
        return true;
    }

    void RowType::printChildren(std::stringstream &ss, std::string_view delimiter)
    const {
        bool any = false;
        for (size_t i = 0; i < children_.size(); ++i) {
            if (any) {
                ss << delimiter;
            }
            const auto &name = names_.at(i);
            if (isColumnNameRequiringEscaping(name)) {
                ss << std::quoted(name, '"', '"');
            } else {
                ss << name;
            }
            ss << ':' << children_.at(i)->toString();
            any = true;
        }
    }

    std::shared_ptr<RowType> RowType::unionWith(
        std::shared_ptr<const RowType> rowType) const {
        std::vector<std::string> names;
        std::vector<TypePtr> types;
        copy(names_.begin(), names_.end(), back_inserter(names));
        copy(rowType->names_.begin(), rowType->names_.end(), back_inserter(names));
        copy(children_.begin(), children_.end(), back_inserter(types));
        copy(
            rowType->children_.begin(),
            rowType->children_.end(),
            back_inserter(types));
        return std::make_shared<RowType>(std::move(names), std::move(types));
    }

    std::string RowType::toString() const {
        std::stringstream ss;
        ss << (TypeTraits<TypeKind::ROW>::name) << "<";
        printChildren(ss);
        ss << ">";
        return ss.str();
    }

    std::optional<int32_t> SerializedTypeCache::get(const Type &type) const {
        auto it = cache_.find(&type);
        if (it != cache_.end()) {
            return it->second.first;
        }

        return std::nullopt;
    }

    int32_t SerializedTypeCache::put(const Type &type, melon::Dynamic serialized) {
        const int32_t id = cache_.size();

        std::pair<int32_t, melon::Dynamic> value{id, std::move(serialized)};
        const bool ok = cache_.emplace(&type, std::move(value)).second;
        POLLUX_CHECK(ok);

        return id;
    }

    melon::Dynamic SerializedTypeCache::serialize() {
        // Make sure to serialize the cache in the same order as it was
        // populated.
        std::vector<std::pair<int32_t, const melon::Dynamic *> > cacheEntries;
        for (const auto &[_, pair]: cache_) {
            cacheEntries.emplace_back(std::make_pair<int32_t, const melon::Dynamic *>(
                (int32_t) pair.first, &pair.second));
        }

        std::sort(cacheEntries.begin(), cacheEntries.end(), [](auto &a, auto &b) {
            return a.first < b.first;
        });

        melon::Dynamic keys = melon::Dynamic::array;
        melon::Dynamic values = melon::Dynamic::array;

        for (const auto &pair: cacheEntries) {
            keys.push_back(pair.first);
            values.push_back(*pair.second);
        }

        melon::Dynamic cacheObj = melon::Dynamic::object;
        cacheObj["keys"] = keys;
        cacheObj["values"] = values;

        return cacheObj;
    }

    SerializedTypeCache &serializedTypeCache() {
        thread_local SerializedTypeCache cache;
        return cache;
    }

    void DeserializedTypeCache::deserialize(const melon::Dynamic &obj) {
        POLLUX_CHECK(cache_.empty());

        const auto &keys = obj["keys"];
        const auto size = keys.size();

        const auto &values = obj["values"];
        POLLUX_CHECK_EQ(size, values.size());
        for (auto i = 0; i < size; ++i) {
            auto type = pollux::ISerializable::deserialize<Type>(values[i]);
            const bool ok = cache_.emplace(keys[i].asInt(), type).second;
            POLLUX_CHECK(ok);
        }
    }

    const TypePtr &DeserializedTypeCache::get(int32_t id) const {
        auto it = cache_.find(id);
        POLLUX_CHECK(it != cache_.end());
        return it->second;
    }

    DeserializedTypeCache &deserializedTypeCache() {
        thread_local DeserializedTypeCache cache;
        return cache;
    }

    namespace {
        melon::Dynamic makeTypeRef(int32_t id) {
            melon::Dynamic ref = melon::Dynamic::object;
            ref["name"] = "Type";
            ref["ref"] = id;
            return ref;
        }
    } // namespace

    melon::Dynamic RowType::serialize() const {
        auto &cache = serializedTypeCache();
        const bool useCache =
                cache.isEnabled() && size() >= cache.options().minRowTypeSize;

        if (useCache) {
            if (auto id = cache.get(*this)) {
                return makeTypeRef(id.value());
            }
        }

        melon::Dynamic obj = melon::Dynamic::object;
        obj["name"] = "Type";
        obj["type"] = TypeTraits<TypeKind::ROW>::name;
        obj["names"] = pollux::ISerializable::serialize(names_);
        obj["cTypes"] = pollux::ISerializable::serialize(children_);

        if (useCache) {
            const auto id = cache.put(*this, std::move(obj));
            return makeTypeRef(id);
        }

        return obj;
    }

    size_t Type::hashKind() const {
        size_t hash = (int32_t) kind() + 1;
        for (auto &child: *this) {
            hash = hash * 31 + child->hashKind();
        }
        return hash;
    }

    bool Type::kindEquals(const TypePtr &other) const {
        // recursive kind match (ignores names)
        if (this->kind() != other->kind()) {
            return false;
        }
        if (this->size() != other->size()) {
            return false;
        }
        for (size_t i = 0; i < this->size(); ++i) {
            if (!this->childAt(i)->kindEquals(other->childAt(i))) {
                return false;
            }
        }
        return true;
    }

    bool MapType::equivalent(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        auto &otherMap = other.as_map();
        return keyType_->equivalent(*otherMap.keyType_) &&
               valueType_->equivalent(*otherMap.valueType_);
    }

    bool MapType::equals(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        auto &otherMap = other.as_map();
        return *keyType_ == *otherMap.keyType_ && *valueType_ == *otherMap.valueType_;
    }

    FunctionType::FunctionType(
        std::vector<std::shared_ptr<const Type> > &&argumentTypes,
        std::shared_ptr<const Type> returnType)
        : children_(allChildren(std::move(argumentTypes), returnType)),
          parameters_{createTypeParameters(children_)} {
    }

    bool FunctionType::equivalent(const Type &other) const {
        if (&other == this) {
            return true;
        }

        if (!Type::hasSameTypeId(other)) {
            return false;
        }

        auto &otherTyped = *reinterpret_cast<const FunctionType *>(&other);
        if (children_.size() != otherTyped.size()) {
            return false;
        }

        for (auto i = 0; i < children_.size(); ++i) {
            if (!children_.at(i)->equivalent(*otherTyped.children_.at(i))) {
                return false;
            }
        }

        return true;
    }

    bool FunctionType::equals(const Type &other) const {
        if (&other == this) {
            return true;
        }

        if (!Type::hasSameTypeId(other)) {
            return false;
        }

        auto &otherTyped = *reinterpret_cast<const FunctionType *>(&other);
        if (children_.size() != otherTyped.size()) {
            return false;
        }

        for (auto i = 0; i < children_.size(); ++i) {
            if (*children_.at(i) != *otherTyped.children_.at(i)) {
                return false;
            }
        }

        return true;
    }

    std::string FunctionType::toString() const {
        std::stringstream out;
        out << "FUNCTION<";
        for (auto i = 0; i < children_.size(); ++i) {
            out << children_[i]->toString() << (i == children_.size() - 1 ? "" : ", ");
        }
        out << ">";
        return out.str();
    }

    melon::Dynamic FunctionType::serialize() const {
        melon::Dynamic obj = melon::Dynamic::object;
        obj["name"] = "Type";
        obj["type"] = TypeTraits<TypeKind::FUNCTION>::name;
        obj["cTypes"] = pollux::ISerializable::serialize(children_);
        return obj;
    }

    OpaqueType::OpaqueType(const std::type_index &typeIndex)
        : typeIndex_(typeIndex) {
    }

    bool OpaqueType::equivalent(const Type &other) const {
        if (&other == this) {
            return true;
        }
        if (!Type::hasSameTypeId(other)) {
            return false;
        }
        return true;
    }

    bool OpaqueType::equals(const Type &other) const {
        if (!this->equivalent(other)) {
            return false;
        }
        auto &otherTyped = *reinterpret_cast<const OpaqueType *>(&other);
        return typeIndex_ == otherTyped.typeIndex_;
    }

    std::string OpaqueType::toString() const {
        std::stringstream out;
        out << "OPAQUE<" << melon::demangle(typeIndex_.name()) << ">";
        return out.str();
    }

    melon::Dynamic OpaqueType::serialize() const {
        const auto &registry = OpaqueSerdeRegistry::get();
        auto it = registry.mapping.find(typeIndex_);
        POLLUX_CHECK(
            it != registry.mapping.end(),
            "No serialization persistent name registered for {}",
            toString());

        melon::Dynamic obj = melon::Dynamic::object;
        obj["name"] = "Type";
        obj["type"] = TypeTraits<TypeKind::OPAQUE>::name;
        obj["opaque"] = it->second.persistentName;
        return obj;
    }

    OpaqueType::SerializeFunc<void> OpaqueType::getSerializeFunc() const {
        const auto &registry = OpaqueSerdeRegistry::get();
        auto it = registry.mapping.find(typeIndex_);
        POLLUX_CHECK(
            it != registry.mapping.end() && it->second.serialize,
            "No serialization function registered for {}",
            toString());
        return it->second.serialize;
    }

    OpaqueType::DeserializeFunc<void> OpaqueType::getDeserializeFunc() const {
        const auto &registry = OpaqueSerdeRegistry::get();
        auto it = registry.mapping.find(typeIndex_);
        POLLUX_CHECK(
            it != registry.mapping.end() && it->second.deserialize,
            "No deserialization function registered for {}",
            toString());
        return it->second.deserialize;
    }

    std::shared_ptr<const OpaqueType> OpaqueType::deserializeExtra(
        const melon::Dynamic &) const {
        return nullptr;
    }

    void OpaqueType::clearSerializationRegistry() {
        auto &registry = OpaqueSerdeRegistry::get();
        registry.mapping.clear();
        registry.reverse.clear();
    }

    void OpaqueType::registerSerializationTypeErased(
        const std::shared_ptr<const OpaqueType> &type,
        const std::string &persistentName,
        SerializeFunc<void> serialize,
        DeserializeFunc<void> deserialize) {
        auto &registry = OpaqueSerdeRegistry::get();
        POLLUX_CHECK(
            !registry.mapping.count(type->typeIndex_),
            "Trying to register duplicated serialization information for type {}",
            type->toString());
        POLLUX_CHECK(
            !registry.reverse.count(persistentName),
            "Trying to register duplicated persistent type name '{}' for type {}, "
            "it's already taken by type {}",
            persistentName,
            type->toString(),
            registry.reverse.at(persistentName)->toString());
        registry.mapping[type->typeIndex_] = {
            .persistentName = persistentName,
            .serialize = serialize,
            .deserialize = deserialize
        };
        registry.reverse[persistentName] = type;
    }

    std::shared_ptr<const ArrayType> ARRAY(TypePtr elementType) {
        return std::make_shared<const ArrayType>(std::move(elementType));
    }

    std::shared_ptr<const RowType> ROW(
        std::vector<std::string> &&names,
        std::vector<TypePtr> &&types) {
        return TypeFactory<TypeKind::ROW>::create(std::move(names), std::move(types));
    }

    std::shared_ptr<const RowType> ROW(
        std::initializer_list<std::pair<const std::string, TypePtr> > &&pairs) {
        std::vector<TypePtr> types;
        std::vector<std::string> names;
        types.reserve(pairs.size());
        names.reserve(pairs.size());
        for (auto &p: pairs) {
            types.push_back(p.second);
            names.push_back(p.first);
        }
        return TypeFactory<TypeKind::ROW>::create(std::move(names), std::move(types));
    }

    std::shared_ptr<const RowType> ROW(std::vector<TypePtr> &&types) {
        std::vector<std::string> names(types.size(), "");
        return TypeFactory<TypeKind::ROW>::create(std::move(names), std::move(types));
    }

    std::shared_ptr<const MapType> MAP(TypePtr keyType, TypePtr valType) {
        return std::make_shared<const MapType>(
            std::move(keyType), std::move(valType));
    }

    std::shared_ptr<const FunctionType> FUNCTION(
        std::vector<TypePtr> &&argumentTypes,
        TypePtr returnType) {
        return std::make_shared<const FunctionType>(
            std::move(argumentTypes), std::move(returnType));
    }

#define POLLUX_DEFINE_SCALAR_ACCESSOR(KIND)                   \
  std::shared_ptr<const ScalarType<TypeKind::KIND>> KIND() { \
    return ScalarType<TypeKind::KIND>::create();             \
  }

    POLLUX_DEFINE_SCALAR_ACCESSOR(INTEGER);
    POLLUX_DEFINE_SCALAR_ACCESSOR(BOOLEAN);
    POLLUX_DEFINE_SCALAR_ACCESSOR(TINYINT);
    POLLUX_DEFINE_SCALAR_ACCESSOR(SMALLINT);
    POLLUX_DEFINE_SCALAR_ACCESSOR(BIGINT);
    POLLUX_DEFINE_SCALAR_ACCESSOR(HUGEINT);
    POLLUX_DEFINE_SCALAR_ACCESSOR(REAL);
    POLLUX_DEFINE_SCALAR_ACCESSOR(DOUBLE);
    POLLUX_DEFINE_SCALAR_ACCESSOR(TIMESTAMP);
    POLLUX_DEFINE_SCALAR_ACCESSOR(VARCHAR);
    POLLUX_DEFINE_SCALAR_ACCESSOR(VARBINARY);

#undef POLLUX_DEFINE_SCALAR_ACCESSOR

    TypePtr UNKNOWN() {
        return TypeFactory<TypeKind::UNKNOWN>::create();
    }

    TypePtr DECIMAL(const uint8_t precision, const uint8_t scale) {
        if (precision <= ShortDecimalType::kMaxPrecision) {
            return std::make_shared<ShortDecimalType>(precision, scale);
        }
        return std::make_shared<LongDecimalType>(precision, scale);
    }

    TypePtr createScalarType(TypeKind kind) {
        return POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(createScalarType, kind);
    }

    TypePtr createType(TypeKind kind, std::vector<TypePtr> &&children) {
        if (kind == TypeKind::FUNCTION) {
            POLLUX_USER_CHECK_GE(
                children.size(),
                1,
                "FUNCTION type should have at least one child type");
            std::vector<TypePtr> argTypes(
                children.begin(), children.begin() + children.size() - 1);
            return std::make_shared<FunctionType>(std::move(argTypes), children.back());
        }

        if (kind == TypeKind::UNKNOWN) {
            POLLUX_USER_CHECK_EQ(
                children.size(), 0, "UNKNOWN type should not have child types");
            return UNKNOWN();
        }
        return POLLUX_DYNAMIC_TYPE_DISPATCH(createType, kind, std::move(children));
    }

    template<>
    TypePtr createType<TypeKind::ROW>(std::vector<TypePtr> && /*children*/) {
        std::string name{TypeTraits<TypeKind::ROW>::name};
        POLLUX_USER_FAIL("Not supported for kind: {}", name);
    }

    template<>
    TypePtr createType<TypeKind::ARRAY>(std::vector<TypePtr> &&children) {
        POLLUX_USER_CHECK_EQ(children.size(), 1, "ARRAY should have only one child");
        return ARRAY(children.at(0));
    }

    template<>
    TypePtr createType<TypeKind::MAP>(std::vector<TypePtr> &&children) {
        POLLUX_USER_CHECK_EQ(children.size(), 2, "MAP should have only two children");
        return MAP(children.at(0), children.at(1));
    }

    template<>
    TypePtr createType<TypeKind::OPAQUE>(std::vector<TypePtr> && /*children*/) {
        std::string name{TypeTraits<TypeKind::OPAQUE>::name};
        POLLUX_USER_FAIL("Not supported for kind: {}", name);
    }

    bool Type::containsUnknown() const {
        if (kind_ == TypeKind::UNKNOWN) {
            return true;
        }
        for (auto i = 0; i < size(); ++i) {
            if (childAt(i)->containsUnknown()) {
                return true;
            }
        }
        return false;
    }

    std::string Type::to_summary_string(TypeSummaryOptions options) const {
        std::ostringstream out;
        out << kindName();

        const auto cnt = std::min(options.maxChildren, size());
        if (cnt > 0) {
            out << "(";
            for (auto i = 0; i < cnt; ++i) {
                if (i > 0) {
                    out << ", ";
                }
                out << childAt(i)->kindName();
            }

            if (cnt < size()) {
                out << ", ..." << (size() - cnt) << " more";
            }
            out << ")";
        } else {
            if (kind_ == TypeKind::ROW) {
                out << "(" << size() << ")";
            }
        }

        return out.str();
    }

    namespace {
        std::unordered_map<std::string, std::unique_ptr<const CustomTypeFactories> > &
        typeFactories() {
            static std::
                    unordered_map<std::string, std::unique_ptr<const CustomTypeFactories> >
                    factories;
            return factories;
        }
    } // namespace

    melon::F14FastMap<std::string, std::type_index> &getTypeIndexByOpaqueAlias() {
        static melon::F14FastMap<std::string, std::type_index>
                typeIndexByOpaqueAlias;
        return typeIndexByOpaqueAlias;
    }

    melon::F14FastMap<std::type_index, std::string> &getOpaqueAliasByTypeIndex() {
        static melon::F14FastMap<std::type_index, std::string>
                opaqueAliasByTypeIndexMap;
        return opaqueAliasByTypeIndexMap;
    }

    bool registerCustomType(
        const std::string &name,
        std::unique_ptr<const CustomTypeFactories> factories) {
        auto uppercaseName = boost::algorithm::to_upper_copy(name);
        return typeFactories().emplace(uppercaseName, std::move(factories)).second;
    }

    bool customTypeExists(const std::string &name) {
        auto uppercaseName = boost::algorithm::to_upper_copy(name);
        return typeFactories().count(uppercaseName) > 0;
    }

    std::unordered_set<std::string> getCustomTypeNames() {
        std::unordered_set<std::string> typeNames;
        for (const auto &[name, unused]: typeFactories()) {
            typeNames.insert(name);
        }
        return typeNames;
    }

    bool unregisterCustomType(const std::string &name) {
        auto uppercaseName = boost::algorithm::to_upper_copy(name);
        return typeFactories().erase(uppercaseName) == 1;
    }

    const CustomTypeFactories * MELON_NULLABLE
    getTypeFactories(const std::string &name) {
        auto uppercaseName = boost::algorithm::to_upper_copy(name);
        auto it = typeFactories().find(uppercaseName);

        if (it != typeFactories().end()) {
            return it->second.get();
        }

        return nullptr;
    }

    TypePtr getCustomType(
        const std::string &name,
        const std::vector<TypeParameter> &parameters) {
        auto factories = getTypeFactories(name);
        if (factories) {
            return factories->getType(parameters);
        }

        return nullptr;
    }

    exec::CastOperatorPtr getCustomTypeCastOperator(const std::string &name) {
        auto factories = getTypeFactories(name);
        if (factories) {
            return factories->getCastOperator();
        }

        return nullptr;
    }

    CustomTypeFactories::~CustomTypeFactories() = default;

    AbstractInputGenerator::~AbstractInputGenerator() = default;

    AbstractInputGeneratorPtr getCustomTypeInputGenerator(
        const std::string &name,
        const InputGeneratorConfig &config) {
        auto factories = getTypeFactories(name);
        if (factories) {
            return factories->getInputGenerator(config);
        }

        return nullptr;
    }

    void toTypeSql(const TypePtr &type, std::ostream &out) {
        switch (type->kind()) {
            case TypeKind::ARRAY:
                // Append <type>[], e.g. bigint[].
                toTypeSql(type->childAt(0), out);
                out << "[]";
                break;
            case TypeKind::MAP:
                // Append map(<key>, <value>), e.g. map(varchar, bigint).
                out << "map(";
                toTypeSql(type->childAt(0), out);
                out << ", ";
                toTypeSql(type->childAt(1), out);
                out << ")";
                break;
            case TypeKind::ROW: {
                // Append struct(name1 type1, name2 type2,..), e.g.
                // struct(a bigint, b real);
                const auto &rowType = type->as_row();
                out << "struct(";
                for (auto i = 0; i < type->size(); ++i) {
                    if (i > 0) {
                        out << ", ";
                    }
                    out << rowType.nameOf(i) << " ";
                    toTypeSql(type->childAt(i), out);
                }
                out << ")";
                break;
            }
            default:
                if (type->isPrimitiveType()) {
                    out << type->toString();
                    return;
                }
                POLLUX_UNSUPPORTED("Type is not supported: {}", type->toString());
        }
    }

    std::string IntervalDayTimeType::valueToString(int64_t value) const {
        static const char *kIntervalFormat = "%s%lld %02d:%02d:%02d.%03d";

        int128_t remainMillis = value;
        std::string sign{};
        if (remainMillis < 0) {
            sign = "-";
            remainMillis = -remainMillis;
        }
        const int64_t days = remainMillis / kMillisInDay;
        remainMillis -= days * kMillisInDay;
        const int64_t hours = remainMillis / kMillisInHour;
        remainMillis -= hours * kMillisInHour;
        const int64_t minutes = remainMillis / kMillisInMinute;
        remainMillis -= minutes * kMillisInMinute;
        const int64_t seconds = remainMillis / kMillisInSecond;
        remainMillis -= seconds * kMillisInSecond;
        char buf[64];
        snprintf(
            buf,
            sizeof(buf),
            kIntervalFormat,
            sign.c_str(),
            days,
            hours,
            minutes,
            seconds,
            remainMillis);

        return buf;
    }

    std::string IntervalYearMonthType::valueToString(int32_t value) const {
        std::ostringstream oss;
        auto sign = "";
        int64_t longValue = value;
        if (longValue < 0) {
            sign = "-";
            longValue = -longValue;
        }
        oss << fmt::format("{}{}-{}", sign, longValue / 12, longValue % 12);
        return oss.str();
    }

    std::string DateType::toString(int32_t days) const {
        return DateType::toIso8601(days);
    }

    std::string DateType::toIso8601(int32_t days) {
        // Find the number of seconds for the days_;
        // Casting 86400 to int64 to handle overflows gracefully.
        int64_t daySeconds = days * (int64_t) (86400);
        std::tm tmValue;
        POLLUX_CHECK(
            Timestamp::epochToCalendarUtc(daySeconds, tmValue),
            "Can't convert days to dates: {}",
            days);
        TimestampToStringOptions options;
        options.mode = TimestampToStringOptions::Mode::kDateOnly;
        // Enable zero-padding for year, to ensure compliance with 'YYYY' format.
        options.zeroPaddingYear = true;
        std::string result;
        result.resize(getMaxStringLength(options));
        const auto view =
                Timestamp::tmToStringView(tmValue, 0, options, result.data());
        result.resize(view.size());
        return result;
    }

    int32_t DateType::toDays(melon::StringPiece in) const {
        return toDays(in.data(), in.size());
    }

    int32_t DateType::toDays(const char *in, size_t len) const {
        return util::fromDateString(in, len, util::ParseMode::kPrestoCast)
                .thenOrThrow(melon::identity, [&](const Status &status) {
                    POLLUX_USER_FAIL("{}", status.message());
                });
    }

    namespace {
        using SingletonTypeMap = melon::F14FastMap<std::string, TypePtr>;

        const SingletonTypeMap &singletonBuiltInTypes() {
            static const SingletonTypeMap kTypes = {
                {"BOOLEAN", BOOLEAN()},
                {"TINYINT", TINYINT()},
                {"SMALLINT", SMALLINT()},
                {"INTEGER", INTEGER()},
                {"BIGINT", BIGINT()},
                {"HUGEINT", HUGEINT()},
                {"REAL", REAL()},
                {"DOUBLE", DOUBLE()},
                {"VARCHAR", VARCHAR()},
                {"VARBINARY", VARBINARY()},
                {"TIMESTAMP", TIMESTAMP()},
                {"INTERVAL DAY TO SECOND", INTERVAL_DAY_TIME()},
                {"INTERVAL YEAR TO MONTH", INTERVAL_YEAR_MONTH()},
                {"DATE", DATE()},
                {"UNKNOWN", UNKNOWN()},
            };
            return kTypes;
        }

        class DecimalParametricType {
        public:
            static TypePtr create(const std::vector<TypeParameter> &parameters) {
                POLLUX_USER_CHECK_EQ(2, parameters.size());
                POLLUX_USER_CHECK(parameters[0].kind == TypeParameterKind::kLongLiteral);
                POLLUX_USER_CHECK(parameters[0].longLiteral.has_value());
                POLLUX_USER_CHECK(parameters[1].kind == TypeParameterKind::kLongLiteral);
                POLLUX_USER_CHECK(parameters[1].longLiteral.has_value());

                return DECIMAL(
                    parameters[0].longLiteral.value(), parameters[1].longLiteral.value());
            }
        };

        class ArrayParametricType {
        public:
            static TypePtr create(const std::vector<TypeParameter> &parameters) {
                POLLUX_USER_CHECK_EQ(1, parameters.size());
                POLLUX_USER_CHECK(parameters[0].kind == TypeParameterKind::kType);
                POLLUX_USER_CHECK_NOT_NULL(parameters[0].type);

                return ARRAY(parameters[0].type);
            }
        };

        class MapParametricType {
        public:
            static TypePtr create(const std::vector<TypeParameter> &parameters) {
                POLLUX_USER_CHECK_EQ(2, parameters.size());
                POLLUX_USER_CHECK(parameters[0].kind == TypeParameterKind::kType);
                POLLUX_USER_CHECK_NOT_NULL(parameters[0].type);

                POLLUX_USER_CHECK(parameters[1].kind == TypeParameterKind::kType);
                POLLUX_USER_CHECK_NOT_NULL(parameters[1].type);

                return MAP(parameters[0].type, parameters[1].type);
            }
        };

        class RowParametricType {
        public:
            static TypePtr create(const std::vector<TypeParameter> &parameters) {
                for (const auto &parameter: parameters) {
                    POLLUX_USER_CHECK(parameter.kind == TypeParameterKind::kType);
                    POLLUX_USER_CHECK_NOT_NULL(parameter.type);
                }

                std::vector<TypePtr> argumentTypes;
                argumentTypes.reserve(parameters.size());
                for (const auto &parameter: parameters) {
                    argumentTypes.push_back(parameter.type);
                }

                return ROW(std::move(argumentTypes));
            }
        };

        class FunctionParametricType {
        public:
            static TypePtr create(const std::vector<TypeParameter> &parameters) {
                POLLUX_USER_CHECK_GE(parameters.size(), 1);
                for (const auto &parameter: parameters) {
                    POLLUX_USER_CHECK(parameter.kind == TypeParameterKind::kType);
                    POLLUX_USER_CHECK_NOT_NULL(parameter.type);
                }

                std::vector<TypePtr> argumentTypes;
                argumentTypes.reserve(parameters.size() - 1);
                for (auto i = 0; i < parameters.size() - 1; ++i) {
                    argumentTypes.push_back(parameters[i].type);
                }

                return FUNCTION(std::move(argumentTypes), parameters.back().type);
            }
        };

        using ParametricTypeMap = melon::F14FastMap<
            std::string,
            std::function<TypePtr(const std::vector<TypeParameter> &parameters)> >;

        const ParametricTypeMap &parametricBuiltinTypes() {
            static const ParametricTypeMap kTypes = {
                {"DECIMAL", DecimalParametricType::create},
                {"ARRAY", ArrayParametricType::create},
                {"MAP", MapParametricType::create},
                {"ROW", RowParametricType::create},
                {"FUNCTION", FunctionParametricType::create},
            };
            return kTypes;
        }
    } // namespace

    bool hasType(const std::string &name) {
        if (singletonBuiltInTypes().count(name)) {
            return true;
        }

        if (parametricBuiltinTypes().count(name)) {
            return true;
        }

        if (customTypeExists(name)) {
            return true;
        }

        return false;
    }

    TypePtr getType(
        const std::string &name,
        const std::vector<TypeParameter> &parameters) {
        if (singletonBuiltInTypes().count(name)) {
            return singletonBuiltInTypes().at(name);
        }

        if (parametricBuiltinTypes().count(name)) {
            return parametricBuiltinTypes().at(name)(parameters);
        }

        return getCustomType(name, parameters);
    }

    std::type_index getTypeIdForOpaqueTypeAlias(const std::string &name) {
        auto it = getTypeIndexByOpaqueAlias().find(name);
        POLLUX_CHECK(
            it != getTypeIndexByOpaqueAlias().end(),
            "Could not find type '{}'. Did you call registerOpaqueType?",
            name);
        return it->second;
    }

    std::string getOpaqueAliasForTypeId(std::type_index typeIndex) {
        auto it = getOpaqueAliasByTypeIndex().find(typeIndex);
        POLLUX_CHECK(
            it != getOpaqueAliasByTypeIndex().end(),
            "Could not find type index '{}'. Did you call registerOpaqueType?",
            typeIndex.name());
        return it->second;
    }
} // namespace kumo::pollux
